#include "mqtt_module.hpp"
#include <glog/logging.h>

yj_mqtt::yj_mqtt(std::string server_url,
                 std::string client_id,
                 std::string username,
                 std::string password,
                 int qos) : _server_url(server_url),
                            _client_id(client_id),
                            _username(username),
                            _password(password),
                            _qos(qos)
{

    _init_flag = YJ_ERROR;
}
yj_mqtt::~yj_mqtt()
{
    _cli->stop_consuming();
    _subscribe_topics.clear();
}

yj_return yj_mqtt::init()
{
    // 创建一个mqtt客户端
    _cli = std::make_shared<mqtt::async_client>(_server_url, _client_id);

    // 创建一个mqtt连接选项
    _conn_opts = mqtt::connect_options_builder::v5()
                     .clean_start(true)
                     .automatic_reconnect(std::chrono::seconds(2), std::chrono::seconds(30))
                     .properties({{mqtt::property::SESSION_EXPIRY_INTERVAL, 604800}})
                     .finalize();

    // 设置用户名和密码
    _conn_opts.set_user_name(_username);
    _conn_opts.set_password(_password);
    print_config();

    _init_flag = YJ_OK;
    LOG(INFO) << "Finish mqtt init";
    return YJ_OK;
}
yj_return yj_mqtt::add_subscribe(std::string topic)
{
    if (topic.empty())
    {
        LOG(ERROR) << "topic is empty";
        return YJ_ERROR;
    }
    _subscribe_topics.push_back(topic);
    return YJ_OK;
}

yj_return yj_mqtt::publish(std::string topic, std::string message, long timeout_ms)
{
    if (_init_flag != YJ_OK)
    {
        LOG(ERROR) << "publish: mqtt is not init";
        return YJ_ERROR;
    }
    if (_cli->is_connected() != true)
    {
        LOG(ERROR) << "publish: mqtt is not connected";
        return YJ_ERROR;
    }

    mqtt::message_ptr pubmsg = mqtt::make_message(topic, message);
    pubmsg->set_qos(_qos);

    if (_cli->publish(pubmsg)->wait_for(timeout_ms) != true)
    {
        LOG(ERROR) << "publish topic:" << topic << " timeout";
        return YJ_ERROR;
    }
    return YJ_OK;
}

yj_return yj_mqtt::start()
{
    while (_init_flag != YJ_OK)
    {
        LOG(INFO) << "start: mqtt is not init, wait for init";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    try
    {
        // Start consumer before connecting to make sure to not miss messages

        _cli->start_consuming();

        // Connect to the server
        LOG(INFO) << "Connecting to the MQTT server...";
        auto tok = _cli->connect(_conn_opts);

        // Getting the connect response will block waiting for the
        // connection to complete.
        auto rsp = tok->get_connect_response();

        // Make sure we were granted a v5 connection.
        if (rsp.get_mqtt_version() < MQTTVERSION_5)
        {
            LOG(ERROR) << "❌ Not a MQTT v5 connection.";
            exit(1);
        }

        // If there is no session present, then we need to subscribe, but if
        // there is a session, then the server remembers us and our
        // subscriptions.
        // if (!rsp.is_session_present())
        // {
        // LOG(INFO) << "Session not present on broker. Subscribing...";
        // LOG(INFO) << "_subscribe_topics.empty():" << _subscribe_topics.empty();
        if (_subscribe_topics.empty())
        {
            LOG(WARNING) << "No topics to subscribe to.";
        }
        else
        {
            for (const auto &topic : _subscribe_topics)
            {
                if (topic != "")
                {
                    LOG(INFO) << "Subscribing to <" << topic << ">...";
                    _cli->subscribe(topic, _qos)->wait();
                }
            }
        }

        // }

        LOG(INFO) << "✔️ Connected to MQTT server and subscribed to topics ok.";
        // Consume messages
        //
        // This just exits if the consumer is closed or the client is
        // disconnected. (See some other examples for auto or manual
        // reconnect)

        LOG(INFO) << "Waiting for messages on topic.";

        try
        {
            while (true)
            {
                auto evt = _cli->consume_event();

                if (const auto *p = evt.get_message_if())
                {
                    auto &msg = *p;
                    if (!msg)
                        continue;

                    while (_recv_buffer.push(msg->to_string()) != YJ_OK) // 当前仅接受一个主题消息，故不做区分
                    {
                        LOG(WARNING) << "⚠️ mqtt cant't push to recv_buffer, retry...";
                        std::this_thread::sleep_for(std::chrono::milliseconds(100));
                    }
                    // LOG(INFO) << msg->get_topic() << ": " << msg->to_string();

                    // const auto &props = msg->get_properties();
                    // if (size_t n = props.size(); n != 0)
                    // {
                    //     std::cout << "\n  [";
                    //     for (size_t i = 0; i < n - 1; ++i)
                    //         std::cout << props[i] << ", ";
                    //     std::cout << props[n - 1] << "]";
                    // }
                    // std::cout << std::endl;
                }
                else if (evt.is_connected())
                {
                    LOG(INFO) << "✔️ Connected";
                }
                else if (evt.is_connection_lost())
                {
                    LOG(ERROR) << "❌ Connection Lost";
                    break;
                }
                else if (const auto *p = evt.get_disconnected_if())
                {
                    LOG(ERROR) << "❌ Disconnected";
                    LOG(ERROR) << " Reason [0x" << std::hex << int{p->reasonCode}
                               << "]: " << p->reasonCode << " ***";
                    break;
                }
            }
        }
        catch (mqtt::queue_closed &exc)
        {
        }

        // If we're here, the client was almost certainly disconnected.
        // But we check, just to make sure.

        if (_cli->is_connected())
        {
            LOG(INFO) << "Shutting down and disconnecting from the MQTT server...";
            _cli->disconnect()->wait();
            LOG(INFO) << "Shutdown complete.";
        }
    }
    catch (const mqtt::exception &exc)
    {
        LOG(INFO) << "Mqtt error code:\n  " << exc;
        return YJ_ERROR;
    }

    return YJ_OK;
}

yj_return yj_mqtt::get_recv_vector(std::vector<std::string> &recv_vector)
{
    return _recv_buffer.move(recv_vector);
}
void yj_mqtt::print_config()
{
    LOG(INFO) << "MQTT url: " << _server_url;
    LOG(INFO) << "MQTT username: " << _username;
    LOG(INFO) << "MQTT password: " << _password;
    LOG(INFO) << "MQTT client_id: " << _client_id;
#ifdef YJ_MQTT_USE_CONFIG_H
    LOG(INFO) << "MQTT subscribe contorl topic: " << YJ_MQTT_SUBSCRIBE_TOPIC_CONTROL;
    LOG(INFO) << "MQTT publish sensor data topic: " << YJ_MQTT_PUBLISH_TOPIC_SENSOR_DATA;
    LOG(INFO) << "MQTT publish status topic: " << YJ_MQTT_PUBLISH_TOPIC_STATUS;
    LOG(INFO) << "MQTT publish control feedback topic: " << YJ_MQTT_PUBLISH_TOPIC_CONTROL_FEEDBACK;
#endif
}